package org.momo.rpc.client.impl;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.momo.rpc.config.RpcConfigLoader;
import org.momo.rpc.client.ChannelHolder;
import org.momo.rpc.client.Client;
import org.momo.rpc.exception.RpcCallException;
import org.momo.rpc.extention.ExtensionLoader;
import org.momo.rpc.protocol.RpcMessage;
import org.momo.rpc.protocol.RpcRequest;
import org.momo.rpc.protocol.codec.NettyDecodeHandler;
import org.momo.rpc.protocol.codec.NettyEncodeHandler;
import org.momo.rpc.registry.ConsumerRegistry;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

@Slf4j
public class NettyClient implements Client {

    private final Bootstrap bootstrap;

    public NettyClient() {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap()
                .group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new IdleStateHandler(10, 50, 0, TimeUnit.SECONDS));
                        pipeline.addLast(new NettyEncodeHandler<RpcMessage<RpcRequest>>());
                        pipeline.addLast(new NettyDecodeHandler());
                        pipeline.addLast(new NettyClientHandler());
                    }
                });
    }

    @Override
    public Channel doConnect(InetSocketAddress serverAddress) throws Exception {
        ChannelFuture cf;
        Channel channel;

        cf = bootstrap.connect(serverAddress).sync();

        if ((channel = cf.channel()) != null) {
            ChannelHolder.putChannel(serverAddress, channel);
        }
        return channel;
    }

    @Override
    public void sendRequest(RpcMessage<RpcRequest> reqMessage, InetSocketAddress serverAddress) {
        RpcMessage.Header reqHeader = reqMessage.getHeader();
        RpcRequest request = reqMessage.getData();

        // get server channel
        Channel channel = ChannelHolder.getChannel(serverAddress);
        if (channel == null) {
            try {
                channel = doConnect(serverAddress);
            } catch (Exception e) {
                String errorMsg = String.format("Client connect error! the error is {}", e.getMessage());
                UnprocessedRequest.remove(reqHeader.getId());
                throw new RpcCallException(reqHeader.getId(), request.getServiceName(), request.getMethodName(), true, errorMsg);
            }
            if(channel == null) {
                UnprocessedRequest.remove(reqHeader.getId());
                throw new RpcCallException(reqHeader.getId(), request.getServiceName(), request.getMethodName(), true, "Unable to Connect!");
            }
        }

        // write to channel
        channel.writeAndFlush(reqMessage).addListener((ChannelFutureListener) channelFuture -> {
            if(!channelFuture.isSuccess()) {
                String errorMsg = String.format("Client send message error! the error is {}", channelFuture.cause().getMessage());
                UnprocessedRequest.remove(reqHeader.getId());
                throw new RpcCallException(reqHeader.getId(), request.getServiceName(), request.getMethodName(), false, errorMsg);
            }
        });

    }

}
